package com.kevin.hw.zklock;

import org.apache.zookeeper.*;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author: kevin
 * @date: 2017/8/14
 * @description:
 */
public class DistributeLock {


    private static final String ROOT_LOCK = "/LOCK";

    private ZooKeeper zookeeper;

    private int sessionTimeout;

    private byte[] lockData = new byte[]{1,2};//节点数据

    private String currentLockId ;//当前的锁节点ID


    private CountDownLatch countDownLatch = new CountDownLatch(1);

    public DistributeLock() throws IOException, InterruptedException {
        this.zookeeper = ZookeeperClient.getInstance();
        this.sessionTimeout = ZookeeperClient.getSessionTimeout();
    }


    public boolean lock(){
        try {

            currentLockId = zookeeper.create(ROOT_LOCK + "/",lockData, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            System.out.println("线程"+Thread.currentThread().getName() + "->成功创建了Lock节点："+currentLockId+"开始去竞争锁");

            List<String> childrenNodes = zookeeper.getChildren(ROOT_LOCK,true);

            //排序 从小到大
            SortedSet<String> sortedSet = new TreeSet<>();
            for (String childeNode : childrenNodes){
                sortedSet.add(ROOT_LOCK+ "/"+childeNode);
            }

            String firstNode = sortedSet.first(); //拿到最小的节点

            //表示当前节点就是最小的节点
            if (currentLockId.equals(firstNode)){

                System.out.println("线程"+Thread.currentThread().getName()+"成功获取锁，LockID: "+firstNode);
                return true;
            }
            SortedSet<String> lessThanLockIds = sortedSet.headSet(currentLockId);

            if (!lessThanLockIds.isEmpty()){
                String lastNode = lessThanLockIds.last();//拿到比当前LOCKID这个几点更小的上一个节点

                //监听上个节点的删除状态
                zookeeper.exists(lastNode, new Watcher() {
                    @Override
                    public void process(WatchedEvent watchedEvent) {
                        if(Event.EventType.NodeDeleted == watchedEvent.getType()){
                            countDownLatch.countDown();
                            System.out.println("监听到节点Lock"+lastNode+"被删除");
                        }
                    }
                });
                countDownLatch.await(sessionTimeout, TimeUnit.MICROSECONDS);

                System.out.println("线程"+Thread.currentThread().getName() + "成功获取锁 ，LockID："+currentLockId);
                return true;
            }

        }catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
        return false;
    }


    public boolean unlock(){
        System.out.println("线程"+Thread.currentThread().getName()+"开始释放锁："+currentLockId);

        try {
            zookeeper.delete(currentLockId,-1);
            System.out.println("节点"+currentLockId+"成功被删除");
            return true;

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
        return false;

    }


}
